package com.teamsun.mqtt.handle;

import io.netty.channel.Channel;
import io.netty.channel.ChannelHandler.Sharable;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import io.netty.handler.codec.mqtt.MqttFixedHeader;
import io.netty.handler.codec.mqtt.MqttMessage;
import io.netty.handler.codec.mqtt.MqttMessageIdVariableHeader;
import io.netty.handler.codec.mqtt.MqttMessageType;
import io.netty.handler.codec.mqtt.MqttQoS;
import io.netty.handler.codec.mqtt.MqttSubAckMessage;
import io.netty.handler.codec.mqtt.MqttSubAckPayload;
import io.netty.handler.codec.mqtt.MqttSubscribeMessage;
import io.netty.handler.codec.mqtt.MqttTopicSubscription;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.LinkedBlockingQueue;


import org.apache.ibatis.session.SqlSession;

import com.teamsun.mqtt.db.DBSessionFactory;
import com.teamsun.mqtt.entity.MsgRep;
import com.teamsun.mqtt.entity.SendMsg;
import com.teamsun.mqtt.service.ChannelDataService;
import com.teamsun.mqtt.service.GetUnReadData;
import com.teamsun.mqtt.service.MessageDataService;

/**
 * 订阅
 * @author acer
 *
 */
@Sharable
public class SubServiceHandle extends  ChannelInboundHandlerAdapter {

	final  String queryMsg="com.teamsun.mqtt.entity.MsgRepMapper.selectByExample";	
	
	
	ConcurrentHashMap<Channel, String> channel2str;
	/**
	 * 消息 队列
	 */
	BlockingQueue<MsgRep> msgRepsque;
	
	/**
	 * 发送过的信息
	 */
	BlockingQueue<SendMsg> msgSendedQue;
	/**
	 * 客户端订阅表 
	  */
	ConcurrentHashMap<String, BlockingQueue<String>> submap;
	
	ExecutorService dboptgroup;

	
	
	public SubServiceHandle(ExecutorService executorService){
		
		msgRepsque=MessageDataService.getMsgRepsque();
		msgSendedQue=MessageDataService.getMsgSendedQue();
		
		submap=MessageDataService.getSubmap();
		channel2str=ChannelDataService.getChannel2str();
		this.dboptgroup=executorService;
		
	}
	
	public void channelRead(ChannelHandlerContext ctx, Object msg) {

		if (msg instanceof MqttMessage) {

			MqttMessage message = (MqttMessage) msg;
			MqttMessageType messageType = message.fixedHeader().messageType();

			switch (messageType) {
			case SUBSCRIBE:
				MqttSubscribeMessage subscribeMessage = (MqttSubscribeMessage) message;
				sub(ctx, subscribeMessage);
				break;
			default:
				ctx.fireChannelRead(msg);
				break;
			}

		}

		else
			ctx.channel().close();
	}
	
	/**
	 * 订阅操作
	 * 向服务器订阅所有感兴趣的主题
	 * 并且按照客户端标识找到未读消息接受这些消息
	 * @param ctx
	 * @param subscribeMessage
	 */
	private void sub(final ChannelHandlerContext ctx,
			MqttSubscribeMessage subscribeMessage) {

	
		MqttFixedHeader fixedHeader = new MqttFixedHeader(
				MqttMessageType.SUBACK, false, MqttQoS.AT_MOST_ONCE, false, 0);

		MqttSubAckPayload payload = new MqttSubAckPayload(2);
		MqttMessageIdVariableHeader mqttMessageIdVariableHeader = MqttMessageIdVariableHeader
				.from(subscribeMessage.variableHeader().messageId());
		MqttSubAckMessage subAckMessage = new MqttSubAckMessage(fixedHeader,
				mqttMessageIdVariableHeader, payload);
		ctx.writeAndFlush(subAckMessage);

		List<MqttTopicSubscription> list = subscribeMessage.payload()
				.topicSubscriptions();

		List<String> topNames = new ArrayList<String>();

		for (MqttTopicSubscription subscription : list) {
			topNames.add(subscription.topicName());
		}
         if(!topNames.isEmpty()){
        	 String iden=channel2str.get(ctx.channel());
        	 
        	 for(String topName:topNames){
        		 BlockingQueue<String> clientIdens=submap.get(topName);
        		 if(clientIdens==null)
        			 clientIdens=new LinkedBlockingQueue<String>();
        		 if(!clientIdens.contains(iden))
        			 clientIdens.offer(iden);
        		 submap.put(topName, clientIdens);
        		 
        	 }

        	 submitGetUnReadyMsg(iden);
         }
		

	}
	
	/**
	 * 在订阅的时候  指定的 client 接受消息
	 * @param clientid
	 */
	private  void  submitGetUnReadyMsg(final String clientid){
		
		 dboptgroup.submit(new  GetUnReadData(clientid));
	}
	
	
	@Override
	public void channelInactive(final ChannelHandlerContext ctx)
			throws Exception {
		super.channelInactive(ctx);
		ctx.close();
	}
}
